/*
 * (C) 2007-2010 Alibaba Group Holding Limited
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * data_entry wrap item of tair, it can be key or value
 *
 * Version: $Id: data_entry.hpp 737 2012-04-17 08:58:19Z ganyu.hfl@taobao.com $
 *
 * Authors:
 *   ruohai <ruohai@taobao.com>
 *     - initial release
 *
 */
#include "data_entry.hpp"

#ifdef WITH_COMPRESS
#include "compressor.hpp"
#include "define.hpp"

int tair::common::data_entry::compress_type = TAIR_SNAPPY_COMPRESS;
int tair::common::data_entry::compress_threshold = TAIR_DEFAULT_COMPRESS_THRESHOLD;
#endif

namespace tair
{
  namespace common {
#ifdef WITH_COMPRESS
    void data_entry::do_compress(tbnet::DataBuffer *output) const
    {
      int new_size = size;
      char *new_data = get_data();
      uint16_t compress_flag = 0;

      if (get_size() > compress_threshold) {
        //do compress process
        char *dest = NULL;
        uint32_t dest_len = 0, src_len = get_size();
        int compress_ret = compressor::do_compress(&dest, &dest_len, get_data(), src_len, compress_type);
        // to check if the ret is valid
        if (0 == compress_ret) {
          new_size = dest_len;
          new_data = dest;
          compress_flag = compress_type;
          compress_flag <<= COMPRESS_TYPE_OFFSET;
          compress_flag |= COMPRESS_FLAG;
        } else {
          log_error("compress error or size overflow, use the raw data instead");
        }
      } else {
        log_debug("data too small, no need to compress");
      }

      // as need to pin two bytes before data
      char *tmp = new char[new_size + TAIR_VALUE_HEADER_LENGTH];
      assert(tmp != NULL);

      tmp[0] = (char) (compress_flag & 0xFF);
      tmp[1] = (char) ((compress_flag >> 8) & 0xFF);

      memcpy(tmp + TAIR_VALUE_HEADER_LENGTH, new_data, new_size);

      if ((0 != compress_flag) && (NULL != new_data)) {
        delete new_data;
      }

      new_data = tmp;
      new_size += TAIR_VALUE_HEADER_LENGTH;

      data_meta.encode(output, true);

      uint32_t msize = (new_size | (prefix_size << PREFIX_KEY_OFFSET));
      output->writeInt32(msize);
      if (get_size() > 0) {
        output->writeBytes(new_data, new_size);
      }

      // if client side, the new_data is alloc while pin the extra
      // two bytes, delete it after encode process
      if (NULL != new_data)
        delete new_data;
    }

    bool data_entry::do_decompress()
    {
      uint16_t compress_header = *(uint16_t *)get_data();
      bool ret = true;

      // data in server side and generated by old client need not to decompress
      if (0 != (data_meta.flag & TAIR_ITEM_FLAG_COMPRESS)) {
        if (0 == (compress_header & COMPRESS_FLAG)) {
          log_debug("uncompressed data");
          int tmp_size = size - TAIR_VALUE_HEADER_LENGTH;
          char *tmp_data = new char[tmp_size];
          memcpy(tmp_data, data + TAIR_VALUE_HEADER_LENGTH, tmp_size);
          free_data();
          set_data(tmp_data, tmp_size, false);
        } else {
          log_debug("compressed data with header: %d", compress_header);
          // do the real compress
          int type = compress_header >> COMPRESS_TYPE_OFFSET;
          char *dest = NULL;
          uint32_t dest_len = 0, src_len = size - TAIR_VALUE_HEADER_LENGTH;
          int compress_ret = compressor::do_decompress(&dest, &dest_len, data + TAIR_VALUE_HEADER_LENGTH, src_len, type);
          // to check if the dest_len overflow
          if (0 == compress_ret) {
            free_data();
            set_data(dest, dest_len, false);
            log_debug("decompress ok, len is %d", dest_len);
          } else {
            ret = false;
            log_error("detect compressed data, but decompress err, code is %d", compress_ret);
          }
        }
      } else {
        log_debug("data generated by old version client");
      }

      return ret;
    }
#endif
  }
}
